package com.fwmagic.spark.core.cases.groupwithtopn

import org.apache.spark.rdd.RDD
import org.apache.spark.{Partitioner, SparkConf, SparkContext}

/**
  * 分组TopN:统计每门学科中访问次数排在前3名的老师
  * 1、对学科和老师组合进行分组统计数量
  * 2、取出前3个
  *
  * 方式二：获取所有学科的种类，根据种类过滤出来
  */
object FavoriteTeacher2 {
    def main(args: Array[String]): Unit = {
        val isLocal = args(0)

        val conf = new SparkConf().setAppName(this.getClass.getSimpleName)

        if (isLocal.toBoolean) {
            conf.setMaster("local[*]")
        }

        val sc = new SparkContext(conf)
        sc.setLogLevel("WARN")
        //读取数据
        val lines: RDD[String] = sc.textFile(args(1))

        //切分数据
        /* http://bigdata.fwmagic.com/huangzhong */
        val subjectTeacherAndOne: RDD[((String, String), Int)] = lines.map(line => {
            val words: Array[String] = line.split("/")
            val subject = words(2).split("\\.")(0)
            val teacher = words(3)
            ((subject, teacher), 1)
        })

        //聚合计算
        val reduced: RDD[((String, String), Int)] = subjectTeacherAndOne.reduceByKey(_ + _)

        //循环中多次调用，可以cache到内存
        reduced.cache() //persist(StorageLevel.MEMORY_ONLY)

        //获取学科的所有种类,收集到Driver端
        val subjects: Array[String] = reduced.map(_._1._1).distinct().collect()

        //取topN
        val topN = args(2).toInt

        //循环、过滤、排序、取TopN
        for (subject <- subjects) {
            val filtered: RDD[((String, String), Int)] = reduced.filter(_._1._1.equals(subject))
            //Driver端调用的是RDD的sortBy，使用的RangePartitioner进行排序（内存+磁盘）
            val tuples: Array[((String, String), Int)] = filtered.sortBy(_._2, false).take(topN)
            println(tuples.toBuffer)
        }

        reduced.unpersist(true)

        sc.stop()
    }
}

/*
预期计算结果：
(javaee,machao,5)
(javaee,sunquan,5)
(javaee,zhangfei,3)
(javaee,sunshangxiang,2)
-- (javaee,caocao,1)

(bigdata,liubei,6)
(bigdata,guanyu,4)
(bigdata,zhaoyun,2)
-- (bigdata,huangzhong,1)

(python,lvbu,4)
(python,zhugeliang,3)
(python,liushan,2)
-- (python,diaochan,1)

准备数据：
http://bigdata.fwmagic.com/huangzhong
http://bigdata.fwmagic.com/zhaoyun
http://bigdata.fwmagic.com/zhaoyun
http://bigdata.fwmagic.com/liubei
http://bigdata.fwmagic.com/liubei
http://bigdata.fwmagic.com/liubei
http://bigdata.fwmagic.com/liubei
http://bigdata.fwmagic.com/liubei
http://bigdata.fwmagic.com/liubei
http://bigdata.fwmagic.com/guanyu
http://bigdata.fwmagic.com/guanyu
http://bigdata.fwmagic.com/guanyu
http://bigdata.fwmagic.com/guanyu
http://javaee.fwmagic.com/zhangfei
http://javaee.fwmagic.com/zhangfei
http://javaee.fwmagic.com/zhangfei
http://javaee.fwmagic.com/machao
http://javaee.fwmagic.com/machao
http://javaee.fwmagic.com/machao
http://javaee.fwmagic.com/machao
http://javaee.fwmagic.com/machao
http://javaee.fwmagic.com/sunquan
http://javaee.fwmagic.com/sunquan
http://javaee.fwmagic.com/sunquan
http://javaee.fwmagic.com/sunquan
http://javaee.fwmagic.com/sunquan
http://javaee.fwmagic.com/caocao
http://javaee.fwmagic.com/sunshangxiang
http://javaee.fwmagic.com/sunshangxiang
http://python.fwmagic.com/lvbu
http://python.fwmagic.com/lvbu
http://python.fwmagic.com/lvbu
http://python.fwmagic.com/lvbu
http://python.fwmagic.com/zhugeliang
http://python.fwmagic.com/zhugeliang
http://python.fwmagic.com/zhugeliang
http://python.fwmagic.com/liushan
http://python.fwmagic.com/liushan
http://python.fwmagic.com/diaochan
*/